package wsmsgsubpub

import (
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/subpub"
	"sync"
)

type WsCliNode struct {
	startedflag  int32
	waitrespflag int8
	startlk      sync.RWMutex
	msgtype      string
	lk           sync.RWMutex
	topicRef     map[string]int
	ex           subpub.ISubPub
	url          string
	subfunc      func(topic string)
	unsubfunc    func(topic string)
	pubfunc      func(topic string, buf []byte) (n int, err error)
	closefunc    func() error
}

func (this *WsCliNode) checkStart() error {
	this.startlk.Lock()
	defer this.startlk.Unlock()
	if this.startedflag == 1 {
		return nil
	}
	defer func() {
		this.startedflag = 1
	}()

	if this.msgtype == "binmsg" {
		cli := NewWsBinMsgCli(this.url)
		cli.WaitRespFlag = this.waitrespflag
		cli.Open()
		this.subfunc = cli.Sub
		this.closefunc = cli.Close
		this.pubfunc = cli.Pub
		this.unsubfunc = cli.UnSub
		go func() {
			if gobase.GoFunCatchException {
				defer gobase.DeferCatchPanic()
			}
			for {
				msg := cli.RecvMsg()
				if msg == nil {
					break
				}
				this.ex.Pub(msg.Topic, 0, msg.Msg)
			}
		}()
	} else {
		cli := NewWsJsonMsgCli(this.url)
		cli.WaitRespFlag = this.waitrespflag
		cli.Open()
		this.subfunc = cli.Sub
		this.closefunc = cli.Close
		this.unsubfunc = cli.UnSub
		this.pubfunc = func(topic string, buf []byte) (n int, err error) {
			return cli.Pub(topic, string(buf))
		}
		go func() {
			if gobase.GoFunCatchException {
				defer gobase.DeferCatchPanic()
			}
			for {
				msg := cli.RecvMsg()
				if msg == nil {
					break
				}
				this.ex.Pub(msg.Topic, 0, msg.Msg)
			}
		}()
	}

	return nil
}

func (this *WsCliNode) Close() error {
	return this.closefunc()
}

func (this *WsCliNode) Unsub(id string, topic string) error {
	this.lk.Lock()
	cnt := this.topicRef[topic]
	cnt--
	if cnt == 0 {
		delete(this.topicRef, topic)
	} else {
		this.topicRef[topic] = cnt
	}
	this.lk.Unlock()
	this.ex.Unsub(id, topic)
	if cnt == 0 {
		this.unsubfunc(topic)
	}
	return nil
}

func (this *WsCliNode) Sub(id string, subtopic string, fn subpub.SubFunc) error {
	this.lk.Lock()
	cnt := this.topicRef[subtopic]
	cnt++
	this.topicRef[subtopic] = cnt
	this.lk.Unlock()

	this.ex.Sub(id, subtopic, fn)
	if cnt == 1 {
		this.subfunc(subtopic)
	}
	return nil
}

func (this *WsCliNode) Pub(topic string, buf []byte) (int, error) {
	return this.pubfunc(topic, buf)
}

var (
	DefaultWsMsgCliGroup = gobase.NewConnectObjectRefStorage(newWsCliNode)
)

func newWsCliNode(connstr string) interface{} {
	conf := gobase.NewStrMap()
	conf.URLFormDecode(connstr)
	node := &WsCliNode{topicRef: make(map[string]int)}
	if conf.Int8ByName("extype", 0) == 0 {
		node.ex = subpub.NewSubchannel()
	} else {
		node.ex = subpub.NewSubscribe()
	}
	node.url = conf.StringByName("url", "")
	node.msgtype = conf.StringByName("msgtype", "binmsg")
	node.waitrespflag = conf.Int8ByName("waitresp", 1)
	return node
}
